package cn._51doit.live.jobs;

import cn._51doit.live.pojo.DataBean;
import cn._51doit.live.udf.IsNewUserFunctionV2;
import cn._51doit.live.udf.JsonToBeanFunction;
import cn._51doit.live.utils.Constants;
import cn._51doit.live.utils.FlinkUtils;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;


/**
 * 使用DataStream API多维度统计新用户
 * 数据中没有isNew这个字段，而是必须实时的计算出该字段，将同一个设备ID的数据，进行keyBy，不论是任何时间的访问数据，一定进入到同一个分区中
 *
 * 1.使用用户的行为数据计算是不是新用户，严格的说不精准，最好使用业务库中的数据进行统计
 * 2.使用布隆过滤器判断用户是否是新用户，有一定的误判率
 * 3.如果按照设备型号进行keyBy，可能会有数据倾斜
 *
 */
public class NewUserCountV2 {


    public static void main(String[] args) throws Exception {

        ParameterTool parameterTool = ParameterTool.fromPropertiesFile(args[0]);

        DataStream<String> lines = FlinkUtils.createKafkaStream(parameterTool, SimpleStringSchema.class);

        SingleOutputStreamOperator<DataBean> beanStream = lines.process(new JsonToBeanFunction());

        //beanStream.print();
        //过滤数据，只要appLaunch类型
        SingleOutputStreamOperator<DataBean> filtered = beanStream.filter(bean -> Constants.APP_LAUNCH.equals(bean.getEventId()));

        //现在使用设备型号进行分区，比例huawei-p50
        KeyedStream<DataBean, String> keyedStream = filtered.keyBy(DataBean::getDeviceType);

        SingleOutputStreamOperator<DataBean> res = keyedStream.process(new IsNewUserFunctionV2());

        //然后将数据写入到ClickHouse中做多维统计


        FlinkUtils.env.execute();


    }
}
